﻿using LJ.EventBus.FastDog.Data;
using LJ.EventBus.FastDog.Transport.Abstractions;
using LJ.EventBus.FastDog.Transport.Abstractions.Exceptions;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace LJ.EventBus.FastDog.Internal;

public class EventBusSubscribe(IEventHandlerManager eventHandlerManager, IMessageSerializer messageSerializer, IConsumerClientFactory consumerClientFactory, IMessageListener messageListener, ILogger<EventBusSubscribe> logger) : IEventBusSubscribe
{
    private readonly TimeSpan _pollingDelay =  TimeSpan.FromSeconds(1);

    private readonly IEventHandlerManager _eventHandlerManager = eventHandlerManager;
    private readonly IMessageSerializer _messageSerializer = messageSerializer;
    private readonly IConsumerClientFactory _consumerClientFactory = consumerClientFactory;
    private readonly IMessageListener _messageListener = messageListener;
    private readonly ILogger _logger = logger;

    public Task Start(CancellationToken stoppingToken)
    {
        // 加载所有的事件处理类
        var eventHandlerDescriptors = _eventHandlerManager.GetEventHandlerDescriptors();

        // 注册事件订阅
        foreach (var eventHandlerDescriptor in eventHandlerDescriptors)
        {
            try
            {
                // ReSharper disable once ConvertToUsingDeclaration
                using var client = _consumerClientFactory.Create(eventHandlerDescriptor.GroupName);
                client.OnLogCallback = WriteLog;
                client.FetchTopics([eventHandlerDescriptor.TopicName]);
            }
            catch (BrokerConnectionException e)
            {
                _logger.LogError(e, "BrokerConnectionException:{Message}", e.Message);
                return Task.CompletedTask;
            }

            _ = Task.Factory.StartNew(()=>
            {
                try
                {
                    // ReSharper disable once ConvertToUsingDeclaration
                    using var client = _consumerClientFactory.Create(eventHandlerDescriptor.GroupName);

                    RegisterMessageProcessor(client, stoppingToken);

                    client.Subscribe([eventHandlerDescriptor.TopicName]);

                    client.Listening(_pollingDelay, stoppingToken);
                }
                catch (OperationCanceledException)
                {
                    //ignore
                }
                catch (BrokerConnectionException e)
                {
                    _logger.LogError(e, "BrokerConnectionException:{Message}" ,e.Message);
                }
                catch (Exception e)
                {
                    _logger.LogError(e, "Exception:{Message}", e.Message);
                }
            }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        }
        return Task.CompletedTask;
    }

    private void RegisterMessageProcessor(IConsumerClient client, CancellationToken cancellationToken)
    {
        client.OnLogCallback = WriteLog;
        client.OnMessageCallback = async (transportMessage, sender) =>
        {
            try
            {
                var name = transportMessage.GetName();

                var eventHandlerDescriptor = _eventHandlerManager.GetEventHandlerDescriptor(name);

                var message = _messageSerializer.Deserialize(transportMessage, eventHandlerDescriptor.EventType);
                _logger.LogDebug("Received message. id:{messageId}, name: {name}", transportMessage.GetId(), name);

                // 执行消息监听处理
                var res = await _messageListener.OnReceiveAsync(message, cancellationToken)
                    .ConfigureAwait(false);
                // 存储偏移,确认消费, see: https://docs.confluent.io/current/clients/dotnet.html
                if (res == MessageReceiveResult.Success)
                {
                    // 只有监听处理成功才提交偏移量,否则不处理即可
                    client.Commit(sender);
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e, "An exception occurred when process received message. Message:'{Message}'.", transportMessage);

                client.Reject(sender);
            }
        };
    }



    private void WriteLog(LogMessageEventArgs logmsg)
    {
        //switch (logmsg.LogType)
        //{
        //    //default:
        //    //    throw new ArgumentOutOfRangeException(nameof(logmsg));
        //}
    }

}
